package com.cmdi.mq.demo1;

import io.netty.channel.DefaultChannelId;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.io.UnsupportedEncodingException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * @Description: 异步发送消息
 * @Author: gaozb
 */
public class AsyncProducer {
    public static void main(String[] args) throws MQClientException, InterruptedException, UnsupportedEncodingException {
        //指定生产者组
        DefaultMQProducer producer = new DefaultMQProducer("yibu-producer-group");
        // 设置NameServer的地址
        producer.setNamesrvAddr("10.254.243.246:9876;10.254.243.249:9876");
        //解决invokeSync call timeout异常
        DefaultChannelId.newInstance();
        // 启动Producer实例
        producer.start();
        // 失败重试次数
        producer.setRetryTimesWhenSendAsyncFailed(0);

        int messageCount = 5;
        // 使用CountDownLatch并发，需要传入参数N
        final CountDownLatch countDownLatch = new CountDownLatch(messageCount);
        for (int i = 0; i < messageCount; i++) {
            try {
                final int index = i;
                //向topic1发送消息。如果没有topic1需要首先创建
                Message msg = new Message("topic1",
                        "TagB",
                        ("Hello world" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
                //设置SendCallback，发送成功或失败的处理方式
                producer.send(msg, new SendCallback() {
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        countDownLatch.countDown();//计算调用次数
                        System.out.printf("%-10d OK %s %n", index, sendResult.getMsgId());
                    }

                    @Override
                    public void onException(Throwable e) {
                        countDownLatch.countDown();//计算调用次数
                        System.out.printf("%-10d Exception %s %n", index, e);
                        e.printStackTrace();
                    }
                });
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        //阻塞线程，直到调用N次countDown()方法才释放
        countDownLatch.await(5, TimeUnit.SECONDS);
        //消息发送完成后，关闭producer服务
        producer.shutdown();
    }
}
